package com.dyava.commons.clock.mysql;


import com.dyava.commons.clock.ClockHandle;
import com.dyava.commons.clock.ClockMessage;
import com.dyava.commons.clock.ClockReceiver;
import com.dyava.commons.lock.KeyLockPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Created by jjliu on 2020/4/17.
 */
public class MysqlClockReceiver extends ClockReceiver implements InitializingBean {
    static Logger log = LoggerFactory.getLogger(MysqlClockReceiver.class);
    ScheduledExecutorService threadPoolClock = Executors.newScheduledThreadPool(1,(r) ->new Thread(r,"thread-mysqlclock"));

    boolean deleteNow=true;

    KeyLockPool keyLockPool;

    List<ClockHandle> clockHandleList = new ArrayList<>();

    JdbcTemplate jdbcTemplate;

    public boolean isDeleteNow() {
        return deleteNow;
    }

    public void setDeleteNow(boolean deleteNow) {
        this.deleteNow = deleteNow;
    }

    public KeyLockPool getKeyLockPool() {
        return keyLockPool;
    }

    public void setKeyLockPool(KeyLockPool keyLockPool) {
        this.keyLockPool = keyLockPool;
    }

    public List<ClockHandle> getClockHandleList() {
        return clockHandleList;
    }

    public void setClockHandleList(List<ClockHandle> clockHandleList) {
        this.clockHandleList = clockHandleList;
    }

    public JdbcTemplate getJdbcTemplate() {
        return jdbcTemplate;
    }

    public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }



    List<MysqlClockRecord> selectClockRecordList(){
        List<MysqlClockRecord> rList = keyLockPool.lock("clock_record",1000,()->{
            try {
                List<MysqlClockRecord> recordList = jdbcTemplate.query(MysqlClockConstant.SQL_SELECT, new RowMapper<MysqlClockRecord>() {
                    @Override
                    public MysqlClockRecord mapRow(ResultSet resultSet, int i) throws SQLException {
                        MysqlClockRecord mysqlClockRecord = new MysqlClockRecord();
                        mysqlClockRecord.setId(resultSet.getLong("id"));
                        mysqlClockRecord.setHandleName(resultSet.getString("handleName"));
                        mysqlClockRecord.setArgs(resultSet.getString("args"));
                        mysqlClockRecord.setState(resultSet.getInt("state"));
                        mysqlClockRecord.setClockTime(resultSet.getTimestamp("clockTime"));
                        mysqlClockRecord.setCreateTime(resultSet.getTimestamp("createTime"));
                        return mysqlClockRecord;
                    }
                });
                if(!recordList.isEmpty()){
                    jdbcTemplate.batchUpdate(MysqlClockConstant.SQL_WORKING,recordList.stream().map((x)->new Long[]{x.getId()}).collect(Collectors.toList()));
                }
                return recordList;
            }catch (Exception e){
                e.printStackTrace();
                return new ArrayList<>();
            }
        });
        return rList;
    }


    protected int roundDispose(){
        int receiverNum=0;
        while(true){
            List<MysqlClockRecord> recordList = selectClockRecordList();
            for(MysqlClockRecord mysqlClockRecord : recordList){
                ClockMessage clockMessage = new ClockMessage(mysqlClockRecord.getHandleName(),mysqlClockRecord.getArgs());
                onReceiver(clockMessage);
            }
            log.trace("-> 搜索{}条",recordList.size());
            if(!recordList.isEmpty()){
                receiverNum+=recordList.size();
                if(deleteNow){
                    jdbcTemplate.batchUpdate(MysqlClockConstant.SQL_DELETE,recordList.stream().map((x)->new Long[]{x.getId()}).collect(Collectors.toList()));
                }else{
                    jdbcTemplate.batchUpdate(MysqlClockConstant.SQL_DONE,recordList.stream().map((x)->new Long[]{x.getId()}).collect(Collectors.toList()));
                }
            }else{
                break;
            }
        }
        return receiverNum;
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        log.info("-> 尝试创建dy_clock_record");
        jdbcTemplate.execute(MysqlClockConstant.SQL_CREATE);
    }

    @Override
    public void start() {
        threadPoolClock.scheduleAtFixedRate(()->{
            roundDispose();
        },1000,1000, TimeUnit.MILLISECONDS);
    }
}
